package com.xjrsoft.common.mqtt;

import cn.hutool.core.lang.TypeReference;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.db.Db;
import cn.hutool.json.JSONUtil;
import com.xjrsoft.common.utils.DatasourceUtil;
import com.xjrsoft.common.utils.IotUtil;
import com.xjrsoft.module.iot.entity.IotDic;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import javax.sql.DataSource;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: tzx
 * @Date: 2024/4/10 14:29
 */
//@Configuration
@Slf4j
public class MqttProviderCallBack implements MqttCallback {

    static List<IotDic> iotDicList;

    static DataSource dataSource;

    private static MqttCallBackModel model;

    static Map<String, Object> cacheChangeMap = new ConcurrentHashMap<>();

    static Map<String, Long> cacheIntervalMap = new ConcurrentHashMap<>();

    public MqttProviderCallBack(MqttCallBackModel myModel) {
        model = myModel;
        iotDicList = IotUtil.initLinkDic(model.getIotLink());
    }

    @Override
    public void connectionLost(Throwable throwable) {
        log.info("连接丢失,原因{}，正在尝试重连！", throwable.getMessage());
        // 连接丢失后，一般在这里面进行重连
        while (!model.getClient().isConnected()) {
            try {
                Thread.sleep(10000);
                model.getClient().reconnect();
                model.getClient().subscribe(model.getIotLink().getTopic(), 2);
                break;
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) {

        try {
            Map<String, Object> payloadMap = JSONUtil.toBean(mqttMessage.toString(), new TypeReference<Map<String, Object>>() {
            }, false);

            //获取到负载json
            List<Map<String, Object>> objs = MapUtil.get(payloadMap, "Objs", new TypeReference<List<Map<String, Object>>>() {
            });

            WebSocketServer.sendInfo(mqttMessage.toString());

            //如果数据字典发生变化 需要更新
            if(IotUtil.isNeedUpdate()){
                iotDicList = IotUtil.initLinkDic(model.getIotLink());
            }

            for (Map<String, Object> obj : objs) {
                String key = MapUtil.get(obj, "N", String.class);
                //获取到当前变量的配置
                Optional<IotDic> currentDic = iotDicList.stream().filter(x -> x.getName().equals(key)).findFirst();
                currentDic.ifPresent(iotDic -> {
                    Object value = MapUtil.get(obj, "1", Object.class);

                    dataSource = DatasourceUtil.getDataSource(iotDic.getSaveDb());

                    if(ObjectUtil.isNotNull(iotDic.getTrans())){
                        value = IotUtil.transValue(iotDic, value);
                    }

                    if (ObjectUtil.isNull(value)) {
                        return;
                    }

                    //不需要保存
                    if (iotDic.getSaveType() == 0) {
                    }
                    //改变的时候 保存
                    else if (iotDic.getSaveType() == 1) {
                        if (!value.equals(cacheChangeMap.get(iotDic.getName()))) {
                            insertData(iotDic, value);
                        }
                    }
                    //间隔  保存
                    else if (iotDic.getSaveType() == 2) {
                        if (cacheIntervalMap.containsKey(iotDic.getName())) {
                            long interval = LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli() - cacheIntervalMap.get(iotDic.getName());
                            if (interval >= Integer.parseInt(iotDic.getSaveValue())) {
                                insertData(iotDic, value);
                            }
                        } else {
                            insertData(iotDic, value);
                            //缓存此次时间
                            cacheIntervalMap.put(iotDic.getName(), LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli());
                        }
                    }
                    // 每次都保存
                    else {
                        insertData(iotDic, value);
                    }
                });
            }

            log.info("主题: " + topic + "   接收到消息：" + mqttMessage);
        } catch (Exception e) {
//            e.printStackTrace();
            log.error(e.getMessage());
        }

    }

    private void insertData(IotDic iotDic, Object value) {
        try {
            String paramSql = "INSERT INTO cy_data (id,var_name,var_type,var_value) VALUES (?,?,?,?);";

            Db.use(dataSource).execute(paramSql, IdUtil.getSnowflakeNextId(), iotDic.getName(), value.getClass().getName(), value);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }



    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

    }
}
